﻿using Mohe.Scada.Config;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Packets;
using NLog;
using System.Text;

namespace Mohe.Scada.Util.Mqtt
{
    public class MqttClusterManager
    {
        private static Logger _logger = LogManager.GetCurrentClassLogger();
        private const string ElectionTopic = "$cluster/election";
        private const string HeartbeatTopic = "$cluster/heartbeat";

        private readonly string _nodeId;
        private bool _isLeader = false;
        private DateTime? _lastHeartbeatTime;
        private Timer _heartbeatTimer;
        private Timer _electionTimer;

        public event EventHandler<bool> LeadershipChanged;

        private readonly IList<IManagedMqttClient> _managedClients;
        private readonly IList<NodeConfig> _brokerUrls;

        public event Func<MqttApplicationMessageReceivedEventArgs, Task> ApplicationMessageReceivedAsync;


        public event Func<MqttClientConnectedEventArgs, Task> ConnectedAsync;

        public bool IsConnected { get => _managedClients.Any(x => x.IsConnected); }

        public MqttClusterManager(IList<NodeConfig> brokerUrls, string nodeId = "")
        {
            _brokerUrls = brokerUrls;
            _nodeId = nodeId;
            _managedClients = new List<IManagedMqttClient>();
        }
        public async Task StartElectionProcess()
        {
            // 订阅选举和心跳主题
            await SubscribeAsync(ElectionTopic);
            await SubscribeAsync(HeartbeatTopic);

            // 设置定时器
            _heartbeatTimer = new Timer(SendHeartbeat, null, TimeSpan.Zero, TimeSpan.FromSeconds(5));
            _electionTimer = new Timer(CheckLeadership, null, TimeSpan.Zero, TimeSpan.FromSeconds(10));


            // 参与选举,匿名节点不参与
            if (!string.IsNullOrEmpty(_nodeId))
            {
                await PublishAsync(ElectionTopic, _nodeId, retain: true);
            }
        }

        private async void SendHeartbeat(object state)
        {
            if (_isLeader)
            {
                await PublishAsync(HeartbeatTopic, _nodeId, retain: true);
            }
        }

        private async void CheckLeadership(object state)
        {
            if (_isLeader) return;

            // 检查最后一次心跳时间
            if (_lastHeartbeatTime.HasValue &&
                (DateTime.UtcNow - _lastHeartbeatTime.Value).TotalSeconds > 15)
            {
                // 心跳超时，发起新的选举
                await PublishAsync(ElectionTopic, _nodeId, retain: true);
            }
        }

        public async Task ConnectAsync()
        {
            foreach (var brokerUrl in _brokerUrls)
            {
                var options = new ManagedMqttClientOptionsBuilder()
                    .WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
                    .WithClientOptions(new MqttClientOptionsBuilder()
                        .WithTcpServer(brokerUrl.Address, brokerUrl.Port)
                        .WithCredentials(brokerUrl.UserName, brokerUrl.Password)
                        .WithClientId($"ClusterManager_{Guid.NewGuid()}")
                        .Build())
                    .Build();

                var managedClient = new MqttFactory().CreateManagedMqttClient();

                // 设置事件处理
                managedClient.ConnectedAsync += OnConnectedAsync;
                managedClient.DisconnectedAsync += OnDisconnectedAsync;
                managedClient.ApplicationMessageReceivedAsync += OnMessageReceivedAsync;

                await managedClient.StartAsync(options);
                _managedClients.Add(managedClient);
            }
        }

        private Task OnConnectedAsync(MqttClientConnectedEventArgs arg)
        {
            return ConnectedAsync?.Invoke(arg);
        }

        private Task OnDisconnectedAsync(MqttClientDisconnectedEventArgs arg)
        {
            _logger.Info($"Disconnected from broker: {arg.ConnectResult?.ResultCode}");
            return Task.CompletedTask;
        }

        private Task OnMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs arg)
        {
            _logger.Info($"Message received on topic {arg.ApplicationMessage.Topic}");
            if (arg.ApplicationMessage.Topic == ElectionTopic)
            {
                HandleElectionMessage(Encoding.UTF8.GetString(arg.ApplicationMessage.PayloadSegment));
            }
            else if (arg.ApplicationMessage.Topic == HeartbeatTopic)
            {
                _lastHeartbeatTime = DateTime.UtcNow;
            }
            // 处理消息逻辑
            return ApplicationMessageReceivedAsync?.Invoke(arg);
        }
        // 发布消息到集群（随机选择一个可用节点）
        public async Task PublishAsync(string topic, object payload, bool retain = false)
        {
            var builder = new MqttApplicationMessageBuilder()
                .WithTopic(topic)
                .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)
                .WithRetainFlag(retain)
                ;
            if (payload is string str)
            {
                builder = builder.WithPayload(str);
            }
            else if (payload is byte[] buf)
            {
                builder = builder.WithPayloadSegment(buf);
            }

            var message = builder.Build();


            try
            {
                var client = _managedClients.FirstOrDefault();
                await client?.EnqueueAsync(message);
            }
            catch (Exception ex)
            {
                _logger.Error($"Publish failed: {ex.Message}");
                // 可以添加重试逻辑或选择其他节点
            }
        }

        // 订阅主题（在所有节点上订阅）
        public async Task SubscribeAsync(params string[] topic)
        {
            var builder = new MqttTopicFilterBuilder();
            foreach (var t in topic)
            {
                builder = builder.WithTopic(t);
            }
            builder.WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce);
            var topicFilter = builder.Build();

            foreach (var client in _managedClients)
            {
                try
                {
                    await client.SubscribeAsync(new List<MqttTopicFilter>() { topicFilter });
                }
                catch (Exception ex)
                {
                    _logger.Error($"Subscribe failed on one node: {ex.Message}");
                }
            }
        }

        // 健康检查
        public Task CheckClusterHealthAsync()
        {
            foreach (var client in _managedClients)
            {
                var status = client.IsConnected ? "Healthy" : "Unhealthy";
                _logger.Info($"Node status: {status}");
            }
            return Task.CompletedTask;
        }
        private void HandleElectionMessage(string candidateId)
        {
            // 简单策略：选择ID最小的节点作为leader
            if (string.Compare(_nodeId, candidateId) <= 0)
            {
                if (!_isLeader)
                {
                    _isLeader = true;
                    LeadershipChanged?.Invoke(this, true);
                    _logger.Info($"I am now the leader: {_nodeId}");
                }
            }
            else
            {
                if (_isLeader)
                {
                    _isLeader = false;
                    LeadershipChanged?.Invoke(this, false);
                    _logger.Info($"I am no longer the leader: {_nodeId}");
                }
            }
        }

        public bool IsLeader => _isLeader;
    }
}
